package com.netflix.client;

import com.netflix.client.config.ClientConfigFactory;
import com.netflix.client.config.CommonClientConfigKey;
import com.netflix.client.config.IClientConfig;
import com.netflix.loadbalancer.ILoadBalancer;
import com.netflix.servo.monitor.Monitors;

import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

import java.lang.reflect.InvocationTargetException;
import java.util.Map;
import java.util.concurrent.ConcurrentHashMap;
import java.util.function.Supplier;

public class ClientFactory {

    private static Map<String, IClient<?, ?>> simpleClientMap = new ConcurrentHashMap<>();
    private static Map<String, ILoadBalancer> namedLBMap = new ConcurrentHashMap<>();
    private static ConcurrentHashMap<String, IClientConfig> namedConfig = new ConcurrentHashMap<>();

    private static Logger logger = LoggerFactory.getLogger(ClientFactory.class);

    public static synchronized IClient<?, ?> registerClientFromProperties(String restClientName, IClientConfig clientConfig) throws ClientException {
        IClient<?, ?> client;
        ILoadBalancer loadBalancer = null;
        if (simpleClientMap.get(restClientName) != null) {
            throw new ClientException(ClientException.ErrorType.GENERAL, "A Rest Client with this name is already registered. Please use a different name");
        }
        try {
            String clientClassName = clientConfig.getOrDefault(CommonClientConfigKey.ClientClassName);
            client = (IClient<?, ?>) instantiateInstanceWithClientConfig(clientClassName, clientConfig);
            boolean initializeNFLoadBalancer = clientConfig.getOrDefault(CommonClientConfigKey.InitializeNFLoadBalancer);
            if (initializeNFLoadBalancer) {
                loadBalancer = registerNamedLoadBalancerFromclientConfig(restClientName, clientConfig);
            }
            if (client instanceof AbstractLoadBalancerAwareClient) {
                ((AbstractLoadBalancerAwareClient) client).setLoadBalancer(loadBalancer);
            }
        } catch (Throwable e) {
            String message = "Unable to InitializeAndAssociateNFLoadBalancer set for RestClient:" + restClientName;
            logger.warn(message, e);
            throw new ClientException(ClientException.ErrorType.CONFIGURATION, message, e);
        }
        simpleClientMap.put(restClientName, client);

        Monitors.registerObject("Client_" + restClientName, client);

        logger.info("Client Registered:" + client.toString());
        return client;
    }

    public static synchronized IClient getNamedClient(String name) {
        if (simpleClientMap.get(name) != null) {
            return simpleClientMap.get(name);
        }
        try {
            return registerClientFromProperties(name, getNamedConfig(name));
        } catch (ClientException e) {
            throw new RuntimeException("Unable to create client", e);
        }
    }

    public static synchronized IClient getNamedClient(String name, Class<? extends IClientConfig> configClass) {
        if (simpleClientMap.get(name) != null) {
            return simpleClientMap.get(name);
        }
        try {
            return createNamedClient(name, configClass);
        } catch (ClientException e) {
            throw new RuntimeException("Unable to create client", e);
        }
    }

    public static synchronized IClient createNamedClient(String name, Class<? extends IClientConfig> configClass) throws ClientException {
        IClientConfig config = getNamedConfig(name, configClass);
        return registerClientFromProperties(name, config);
    }

    public static synchronized ILoadBalancer getNamedLoadBalancer(String name) {
        ILoadBalancer lb = namedLBMap.get(name);
        if (lb != null) {
            return lb;
        } else {
            try {
                lb = registerNamedLoadBalancerFromclientConfig(name, getNamedConfig(name));
            } catch (ClientException e) {
                throw new RuntimeException("Unable to create load balancer", e);
            }
            namedLBMap.put(name, lb);
            return lb;
        }
    }

    public static synchronized ILoadBalancer getNamedLoadBalancer(String name, Class<? extends IClientConfig> configClass) {
        ILoadBalancer lb = namedLBMap.get(name);
        if (lb != null) {
            return lb;
        } else {
            try {
                lb = registerNamedLoadBalancerFromProperties(name, configClass);
            } catch (ClientException e) {
                throw new RuntimeException("Unable to create load balancer", e);
            }
            return lb;
        }
    }

    public static ILoadBalancer registerNamedLoadBalancerFromclientConfig(String name, IClientConfig clientConfig) throws ClientException {
        if (namedLBMap.get(name) != null) {
            throw new ClientException("LoadBalancer for name " + name + " already exists");
        }
        try {
            String loadBalancerClassName = clientConfig.getOrDefault(CommonClientConfigKey.NFLoadBalancerClassName);
            ILoadBalancer lb = (ILoadBalancer) ClientFactory.instantiateInstanceWithClientConfig(loadBalancerClassName, clientConfig);
            namedLBMap.put(name, lb);
            logger.info("Client: {} instantiated a LoadBalancer: {}", name, lb);
            return lb;
        } catch (Throwable e) {
            throw new ClientException("Unable to instantiate/associate LoadBalancer with Client:" + name, e);
        }
    }

    public static synchronized ILoadBalancer registerNamedLoadBalancerFromProperties(String name, Class<? extends IClientConfig> configClass) throws ClientException {
        if (namedLBMap.get(name) != null) {
            throw new ClientException("LoadBalancer for name " + name + " already exists");
        }
        IClientConfig clientConfig = getNamedConfig(name, configClass);
        return registerNamedLoadBalancerFromclientConfig(name, clientConfig);
    }

    @SuppressWarnings("unchecked")
    public static Object instantiateInstanceWithClientConfig(String className, IClientConfig clientConfig) throws InstantiationException, IllegalAccessException, ClassNotFoundException {
        Class clazz = Class.forName(className);
        if (IClientConfigAware.class.isAssignableFrom(clazz)) {
            IClientConfigAware obj = (IClientConfigAware) clazz.newInstance();
            obj.initWithNiwsConfig(clientConfig);
            return obj;
        } else {
            try {
                if (clazz.getConstructor(IClientConfig.class) != null) {
                    return clazz.getConstructor(IClientConfig.class).newInstance(clientConfig);
                }
            } catch (NoSuchMethodException ignored) {
                // OK for a class to not take an IClientConfig
            } catch (SecurityException | IllegalArgumentException | InvocationTargetException e) {
                logger.warn("Error getting/invoking IClientConfig constructor of {}", className, e);
            }
        }
        logger.warn("Class " + className + " neither implements IClientConfigAware nor provides a constructor with IClientConfig as the parameter. Only default constructor will be used.");
        return clazz.newInstance();
    }

    public static IClientConfig getNamedConfig(String name) {
        return getNamedConfig(name, ClientConfigFactory.DEFAULT::newConfig);
    }

    public static IClientConfig getNamedConfig(String name, Class<? extends IClientConfig> clientConfigClass) {
        return getNamedConfig(name, factoryFromConfigType(clientConfigClass));
    }

    public static IClientConfig getNamedConfig(String name, Supplier<IClientConfig> factory) {
        return namedConfig.computeIfAbsent(name, ignore -> {
            try {
                IClientConfig config = factory.get();
                config.loadProperties(name);
                return config;
            } catch (Exception e) {
                logger.error("Unable to create named client config '{}' instance for config factory {}", name, factory, e);
                return null;
            }
        });
    }

    private static Supplier<IClientConfig> factoryFromConfigType(Class<? extends IClientConfig> clientConfigClass) {
        return () -> {
            try {
                return (IClientConfig) clientConfigClass.newInstance();
            } catch (Exception e) {
                throw new RuntimeException(String.format("Failed to create config for class '%s'", clientConfigClass));
            }
        };
    }
}
